package com.pro

import com.`trait`._
import com.utils.{ConfigUtils, DateUtils, KuduUtils}
import org.apache.kudu.client.CreateTableOptions
import org.apache.kudu.spark.kudu.KuduContext
import org.apache.spark.sql.SparkSession

object AdRegionAnalysis extends Process {
  override def process(spark: SparkSession, kuduContext: KuduContext): Unit = {

    val options = Map(
      "kudu.master" -> ConfigUtils.KUDU_MASTER,
      "kudu.table" -> "ODS_20190822")
    import org.apache.kudu.spark.kudu._
    spark.read.options(options).kudu.createOrReplaceTempView("t_table_info")

    spark.sparkContext.setLogLevel("warn")

    val result = spark.sql(
      """
        | select proviceName,city,org_num,valid_num,ad_num,bid_num,bid_success_num,
        | case when isnull(bid_success_num/bid_num) then 0 else bid_success_num/bid_num end as bid_success_rat,
        | show_num,click_num,
        | case when isnull(click_num/show_num) then 0 else click_num/show_num end as click_rat,
        | media_show_num,media_click_num,consumtion_number,cost_number
        | from (
        | select proviceName,city,
        | sum(case when requestmode==1 and processnode>=1 then 1 else 0 end) org_num,
        | sum(case when requestmode=1 and processnode>=2 then 1 else 0 end) valid_num,
        | sum(case when requestmode=1 and processnode=3 then 1 else 0 end) ad_num,
        | sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and isbid=1
        | and adorderid!=0 then 1 else 0 end) bid_num,
        | sum(case when iseffective=1 and iswin=1 and isbilling=1 and adplatformproviderid>=100000
        | then 1 else 0 end) bid_success_num,
        | sum(case when requestmode=2 and iseffective=1 then 1 else 0 end) show_num,
        | sum(case when requestmode=3 and iseffective=1 then 1 else 0 end) click_num,
        | sum(case when requestmode=2 and iseffective=1 and isbilling=1 then 1 else 0 end) media_show_num,
        | sum(case when requestmode=3 and iseffective=1 and isbilling=1 then 1 else 0 end) media_click_num,
        | sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and
        | iswin=1 and adorderid>200000 and adcreativeid>200000 then winprice/1000 else 0 end) consumtion_number,
        | sum(case when adplatformproviderid>=100000 and iseffective=1 and isbilling=1 and
        | iswin=1 and adorderid>200000 and adcreativeid>200000 then adpayment/1000 else 0 end) cost_number
        |from t_table_info group by proviceName,city ) a
      """.stripMargin)
    val option = new CreateTableOptions()
    val columns = Seq[String]("proviceName","city")
    import scala.collection.JavaConverters._
    option.addHashPartitions(columns.asJava,3)
    option.setNumReplicas(3)

    KuduUtils.writeToKudu(kuduContext,result.schema,option,s"ad_region_analysis_${DateUtils.getNow()}",columns,result)

  }
}
